/*
 * Project: push-center-service-impl
 * 
 * File Created at 2014-10-9
 * 
 * Copyright 2012 Greenline.com Corporation Limited.
 * All rights reserved.
 *
 * This software is the confidential and proprietary information of
 * Greenline Company. ("Confidential Information").  You shall not
 * disclose such Confidential Information and shall use it only in
 * accordance with the terms of the license agreement you entered into
 * with Greenline.com.
 */
package com.greenline.push.service.core;

import java.util.ArrayList;
import java.util.List;

import javax.annotation.Resource;

import org.apache.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;

import com.greenline.push.core.enums.Channels;
import com.greenline.push.core.enums.MsgSendType;
import com.greenline.push.core.pushcenter.MsgForSendManager;
import com.greenline.push.core.pushcenter.MsgHisManager;
import com.greenline.push.dal.pushcenter.dataobject.MsgForSendDO;
import com.greenline.push.service.core.counter.CounterManager;
import com.greenline.push.service.core.fetch.FetcherManager;
import com.greenline.push.service.core.fetch.IAsyncFetcher;
import com.greenline.push.service.core.fetch.IFetcher;
import com.greenline.push.service.core.fetch.ISyncFetcher;
import com.greenline.push.service.core.item.Packet;
import com.greenline.push.service.core.recv.IReceiver;
import com.greenline.push.service.core.send.base.ISender;
import com.greenline.push.service.core.send.base.SenderManager;
import com.greenline.push.service.core.tmpl.AppChannelCache;
import com.greenline.push.service.core.tmpl.AppTemplateCache;
import com.greenline.push.service.core.tmpl.AppTemplateInfo;
import com.greenline.push.service.core.utils.Constants;

/**
 * 消息主处理器
 * 
 * @author weirui.shenwr
 * @date 2014-10-9
 * @version V1.0
 */
public class DefaultProcessor implements IProcessor {

    // private Log log = LogFactory.getLog(DefaultProcessor.class);
    private Logger log = Constants.LOG_PROCESS;

    private boolean isStart = false;

    private final int process_queue_count = 5;
    private final int process_queue_max_remain_size = 10000;
    private final int process_queue_listener_count = 2;

    private ProcessQueueList processQueueList;
    @Resource
    private IReceiver receiver;

    @Autowired
    private FetcherManager fetchManager;

    @Resource
    private AppChannelCache appChannelCache;

    @Resource
    private SenderManager senderManager;

    // 同步获取器最多只能有一个
    private ISyncFetcher syncFetcher;

    private List<ProcessQueueListener> threads = null;
    @Resource
    private MsgHisManager msgHis;

    @Autowired
    private MsgForSendManager msgForSendManager;
    @Autowired
    private CounterManager counterManager;
    @Autowired
    private AppTemplateCache templateCache;

    /**
     * 初始化方法
     */
    public synchronized void start() {
        log.info("DefaultProcessor处理器初始化start=========");
        if (!isStart) {
            // 初始化队列
            processQueueList = new ProcessQueueList(process_queue_count, process_queue_max_remain_size);
            counterManager.addQueue("总接收队列", processQueueList);
            // start fetcher (异步获取消息器，会有定时获取器，在没有同步获取器的时候还有通过中间存储的获取器）
            for (IFetcher fetcher : fetchManager.getFetcherList()) {
                if (fetcher instanceof IAsyncFetcher) {
                    ((IAsyncFetcher) fetcher).fetch(processQueueList);
                } else if (fetcher instanceof ISyncFetcher) {
                    syncFetcher = (ISyncFetcher) fetcher;
                }
            }

            // 启动线程
            threads = new ArrayList<DefaultProcessor.ProcessQueueListener>(process_queue_count * process_queue_listener_count);
            for (int i = 0; i < process_queue_count; i++) {
                for (int j = 0; j < process_queue_listener_count; j++) {
                    ProcessQueueListener listener = new ProcessQueueListener(i, processQueueList);
                    listener.setName("[消息接收处理器]queue-processor-thread" + (i * process_queue_listener_count + j));
                    listener.start();
                    threads.add(listener);
                }
            }
            isStart = true;
        }
        log.info("DefaultProcessor处理器初始化end=========");
    }

    public void stop() {
        try {
            for (DefaultProcessor.ProcessQueueListener listener : threads) {
                listener.threadStop();
                listener.interrupt();
                while (listener.isAlive()) {
                    Thread.sleep(100);
                }
            }
        } catch (InterruptedException e) {
            threads = null;
            log.error(e);
        }
    }

    @Override
    public boolean process(Packet packet) {
        log.debug(String.format("process收到消息:类型%s,接受者%s,telplateId:%s,domainId:%s", packet.getType(), packet.getRecvUserId(),
                packet.getTemplateId(), packet.getDomainId()));

        // this type is handled by task.here only fetch to db
        if (MsgSendType.getMsgType(packet.getType()).isSkipQueue()) {
            return receiver.receive(packet);
        }

        // 判断是定时任务还是实时任务
        Long sendingTime = packet.getSendingTime();
        if (sendingTime != null) {
            // this type onley receive to db.the task handle
            receiver.receive(packet);
        } else {
            if (syncFetcher != null) {
                receiver.receive(packet);
                boolean flag = syncFetcher.fetch(processQueueList, packet);
                if (!flag) {
                    // 入队列失败，回头更新数据库的fetch_count 为0 表示首次入队失败
                    MsgForSendDO msgForSendDO = msgForSendManager.getById(packet.getDbMsgId());
                    msgForSendDO.setFetchCount(0);
                    msgForSendManager.update(msgForSendDO);
                    log.info("{msg_for_send入队列}失败,更新数据库为失败状态。id是:" + msgForSendDO.getId());
                }
            }
        }
        return true;
    }

    private void processPush(Packet packet) {

        // the list has been sorted in cache.
        // find by domain
        List<AppTemplateInfo> appTmpls = templateCache.get(packet.getTemplateId(), packet.getDomainId());
        if (null == appTmpls) {
            log.error("该域下未找到对应的模板:domain:" + packet.getDomainId());
            return;
        }

        int priority = -1;
        for (AppTemplateInfo appTempl : appTmpls) {
            // 优先级比上次处理的大（第一次或者前次推送失败）或者相等就调用推送
            if (appTempl.getPriority() >= priority) {
                Channels channel = appChannelCache.getChannel(appTempl.getAppChannelId());
                // 调用发送
                ISender sender = senderManager.get(channel);
                // 入队列成功，则返回true，否则返回false
                if (sender.send(packet, appTempl)) {
                    priority = appTempl.getPriority();
                } else {
                    log.info(String.format("{processPush}.推送%s失败,选择低优先级(如果有的话).", channel));
                }
            }
        }
    }

    /**
     * 
     * ProcessQueueListener
     * 
     * @author weirui.shenwr
     * @date 2014-10-10
     * @version V1.0
     */
    private class ProcessQueueListener extends Thread {

        private int queueIndex;
        private boolean threadStopped = false;
        ProcessQueueList queueList;

        public ProcessQueueListener(int queueIndex, ProcessQueueList queueList) {
            this.queueIndex = queueIndex;
            this.queueList = queueList;
        }

        @Override
        public void run() {
            while (!threadStopped) {

                Packet packet = null;

                try {
                    // 读取待推送队列的包
                    packet = queueList.take(queueIndex);
                    // 调用推送处理
                    processPush(packet);
                } catch (Exception e) {
                    log.error("[" + getName() + "] Exception during packet processing: " + packet, e);
                }
            }
        }

        public void threadStop() {
            threadStopped = true;
        }
    }
}
